package com.zhu.app.dwd;

import com.zhu.utils.ZhuKafkaUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 *
 * 读取用户表数据，获取注册时间，将用户注册信息写入 Kafka 用户注册主题。
 * 9.16.2 思路分析
 * 用户注册时会在用户表中插入一条数据，筛选操作类型为 insert 的数据即可。
 */
public class DWDUserRegisterApp {
    public static void main(String[] args) throws Exception {

        //todo env
        StreamExecutionEnvironment streamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment();
        streamExecutionEnvironment.setParallelism(4); //  kafka partition 4

        //check point
           /*
        streamExecutionEnvironment.setStateBackend(new HashMapStateBackend());
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointStorage(ClusterParametersConfig.HDFS_CHECKPOINT_FILE_DIR);  //检查点保存在hdfs
        System.setProperty("HADOOP_USER_NAME", "zhu");
        streamExecutionEnvironment.getCheckpointConfig().setCheckpointTimeout(10 * 60000L);  //TimeOut
        streamExecutionEnvironment.getCheckpointConfig().setMaxConcurrentCheckpoints(2);  //最大共存检查点
        streamExecutionEnvironment.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5 * 1000L));  //重启策略
        */
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(streamExecutionEnvironment);
        Configuration configuration = streamTableEnvironment.getConfig().getConfiguration();


        //todo kafka topic_db
        streamTableEnvironment.executeSql(ZhuKafkaUtil.getTopicDB("dwd_user_region"));


        // TODO 4. 读取用户表数据
        Table userInfo = streamTableEnvironment.sqlQuery("select\n" +
                "data['id'] user_id,\n" +
                "data['create_time'] create_time, \n" +
                "ts " +
                "from topic_db\n" +
                "where `table` = 'user_info'\n" +
                "and `type` = 'insert'\n");
        streamTableEnvironment.createTemporaryView("user_info", userInfo);

        streamTableEnvironment.executeSql("create table `dwd_user_register`(\n" +
                "`user_id` string,\n" +
                "`date_id` string,\n" +
                "`create_time` string, \n" +
                "ts string " +
                ")" +
                ZhuKafkaUtil.getKafkaSinkDDL("dwd_user_register"));
// TODO 6. 将输入写入 Kafka-Connector 表
        streamTableEnvironment.executeSql("insert into dwd_user_register\n" +
                "select \n" +
                "user_id,\n" +
                "date_format(create_time, 'yyyy-MM-dd') date_id,\n" +
                "create_time, \n" +
                "ts " +
                "from user_info").print();


        streamExecutionEnvironment.execute();
    }
}
